package com.atguigu.flink.state;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/3/1
 *
 *  keyed state 和Opearator state的区别。
 *          Operator state: 是一个task ，维护一个状态，所有的key都共用一个状态。
 *          keyed state:   是一个task 收到的每一个key 维护一个状态，不同的key用各自的状态，互补影响。
 *                          必须keyBy
 *                          keyBy之后流就是 KeyedStream,拥有RichFunction中的方法。
 *                              可以使用 RichFunction中的生命周期方法，在Task创建时，从备份中恢复状态。
 *
 * 检测每种传感器的水位值，如果连续的两个水位值超过10，就输出报警
 */
public class Demo4_KeyedValueState
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        
                env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   .keyBy(WaterSensor::getId)
                   .process(new KeyedProcessFunction<String, WaterSensor, String>()
                   {

                       @Override
                       public void open(Configuration parameters) throws Exception {
                           //在task启动时，从备份中恢复状态
                           lastVc = getRuntimeContext().getState(new ValueStateDescriptor<>("lastVc", Integer.class));
                       }

                       //声明一个变量保存之前的vc
                       //int lastVc = 0;
                       //声明一个ValueState，用于保存单个值(任意对象，String，整数)
                       private ValueState<Integer> lastVc;
                       @Override
                       public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                           //取出 ValueState 存储的值
                           Integer lastVcNum = lastVc.value();

                           if (lastVcNum != null && lastVcNum > 10 && value.getVc() > 10){
                               out.collect(ctx.getCurrentKey() +" 连续的两个水位值超过10，报警!");
                           }

                           //更新lastVc  update(): 覆盖写
                           lastVc.update(value.getVc());

                       }
                   })
                   .print();

        
                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
        
    }
}
